/*
 * Copyright 2020-2030 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      https://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.wwl.config;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DefaultConsumer;
import com.wwl.repository.RabbitCustomProperties;
import com.wwl.repository.RabbitExchangeDefinition;
import com.wwl.repository.RabbitQueueDefinition;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;

import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeoutException;

/**
 * @author wangweili
 * @version 1.0
 * @date 2021/3/23 8:54 下午
 */
@Slf4j
@Configuration
@Import({RabbitMqRegister.class, RabbitTemplate.class})
@ConditionalOnClass(RabbitMqConfig.class)
@EnableConfigurationProperties(RabbitCustomProperties.class)
@SuppressWarnings("all")
public class RabbitMqConfig {

    private static final String DELAY = "_delay";

    private static final String EXCHANGE_LOGS = "声明交换机：{}类型：{}";

    private static final String QUEUE_LOGS = "队列：{}绑定交换机：{}路由：{}";

    private final RabbitCustomProperties rabbitCustomProperties;

    private final RabbitMqRegister rabbitMqRegister;

    public RabbitMqConfig(RabbitCustomProperties rabbitCustomProperties, RabbitMqRegister rabbitMqRegister) {
        this.rabbitCustomProperties = rabbitCustomProperties;
        this.rabbitMqRegister = rabbitMqRegister;
    }

    public void registy() {
        try {
            createMqExchange(getRabbitDefinition());
            declareQueueAndBinding(getRabbitDefinition());
        } catch (Exception e) {
            log.error(e.getMessage());
        }
    }

    public void createListener(List<DefaultConsumer> consumers) {
        try {
            initListeners(consumers, this.rabbitCustomProperties);
        } catch (Exception e) {
            log.error(e.getMessage());
        }
    }

    private void exchangeDirect(String exchangeName, Boolean delayExchange, String exchangeType) {
        log.info(EXCHANGE_LOGS, exchangeName, exchangeType);
        rabbitMqRegister.registerBean(
                exchangeName,
                ExchangeBuilder.directExchange(exchangeName).durable(true).build());
        if (delayExchange) {
            rabbitMqRegister.registerBean(
                    exchangeName + DELAY,
                    ExchangeBuilder.directExchange(exchangeName + DELAY).durable(true).build());
        }
    }

    private void exchangeFanout(String exchangeName, Boolean delayExchange, String exchangeType) {
        log.info(EXCHANGE_LOGS, exchangeName, exchangeType);
        rabbitMqRegister.registerBean(
                exchangeName,
                ExchangeBuilder.fanoutExchange(exchangeName).durable(true).build());
    }

    private void exchangeTopic(String exchangeName, Boolean delayExchange, String exchangeType) {
        log.info(EXCHANGE_LOGS, exchangeName, exchangeType);
        rabbitMqRegister.registerBean(
                exchangeName,
                ExchangeBuilder.topicExchange(exchangeName).durable(true).build());
    }

    private void queueDirect(List<RabbitQueueDefinition> rabbitQueueDefinitions, String exchangeName, Boolean delayExchange) {
        rabbitQueueDefinitions.forEach(queueDefinition -> {
            String queueName = queueDefinition.getQueueName();
            String routingKey = queueDefinition.getRoutingKey();
            log.info(QUEUE_LOGS, queueName, exchangeName, routingKey);
            declareQueue(queueName, delayExchange, exchangeName, routingKey);
            String bindingBean = queueName + "." + exchangeName;
            rabbitMqRegister.registerBean(
                    bindingBean,
                    BindingBuilder.bind(rabbitMqRegister.getBean(queueName, Queue.class))
                            .to(rabbitMqRegister.getBean(exchangeName, DirectExchange.class))
                            .with(routingKey));
            if (delayExchange) {
                declareQueue(queueName + DELAY);
                bindingBean = queueName + "." + exchangeName + DELAY;
                log.info(QUEUE_LOGS, queueName + DELAY, exchangeName + DELAY, routingKey + DELAY);
                //死信DIRECT交换机绑定死信队列
                rabbitMqRegister.registerBean(
                        bindingBean,
                        BindingBuilder.bind(rabbitMqRegister.getBean(queueName + DELAY, Queue.class))
                                .to(rabbitMqRegister.getBean(exchangeName + DELAY, DirectExchange.class))
                                .with(routingKey + DELAY));
            }
        });
    }

    private void queueFanout(List<RabbitQueueDefinition> rabbitQueueDefinitions, String exchangeName, Boolean delayExchange) {
        rabbitQueueDefinitions.forEach(queueDefinition -> {
            String queueName = queueDefinition.getQueueName();
            String routingKey = queueDefinition.getRoutingKey();
            log.info(QUEUE_LOGS, queueName, exchangeName, routingKey);
            declareQueue(queueName);
            String bindingBean = queueName + "." + exchangeName;
            rabbitMqRegister.registerBean(
                    bindingBean,
                    BindingBuilder.bind(rabbitMqRegister.getBean(queueName, Queue.class))
                            .to(rabbitMqRegister.getBean(exchangeName, FanoutExchange.class)));
        });
    }

    private void queueTopic(List<RabbitQueueDefinition> rabbitQueueDefinitions, String exchangeName, Boolean delayExchange) {
        rabbitQueueDefinitions.forEach(queueDefinition -> {
            String queueName = queueDefinition.getQueueName();
            String routingKey = queueDefinition.getRoutingKey();
            log.info(QUEUE_LOGS, queueName, exchangeName, routingKey);
            declareQueue(queueName);
            String bindingBean = queueName + "." + exchangeName;
            rabbitMqRegister.registerBean(
                    bindingBean,
                    BindingBuilder.bind(rabbitMqRegister.getBean(queueName, Queue.class))
                            .to(rabbitMqRegister.getBean(exchangeName, TopicExchange.class)).with(routingKey));
        });

    }

    private void initListeners(List<DefaultConsumer> consumers, RabbitCustomProperties customProperties) throws IOException, TimeoutException {
        List<RabbitExchangeDefinition> rabbitDefinitions = customProperties.getRabbitDefinition();
        doCreate(consumers, customProperties, rabbitDefinitions);
    }

    private void doCreate(List<DefaultConsumer> consumers, RabbitCustomProperties customProperties, List<RabbitExchangeDefinition> rabbitDefinitions) throws IOException, TimeoutException {
        for (RabbitExchangeDefinition definition : rabbitDefinitions) {
            String exchangeName = definition.getExchangeName();
            String exchangeType = definition.getExchangeType();
            Boolean delayExchange = definition.getDelayExchange();
            List<RabbitQueueDefinition> queueDefinitions = definition.getQueueDefinitions();
            for (RabbitQueueDefinition queueDefinition : queueDefinitions) {
                Channel channel = getChannel(customProperties);
                declareListener(consumers, exchangeName, queueDefinition, channel);
            }
        }
    }

    private void declareListener(List<DefaultConsumer> consumers, String exchangeName, RabbitQueueDefinition queueDefinition, Channel channel) throws IOException {
        channel.queueDeclare(queueDefinition.getQueueName(), true, false, false, null);
        channel.queueBind(queueDefinition.getQueueName(), exchangeName, queueDefinition.getRoutingKey());
        DefaultConsumer consumer = consumers.remove(0);
        channel.basicConsume(queueDefinition.getQueueName(), true, consumer);
    }

    private Channel getChannel(RabbitCustomProperties customProperties) throws IOException, TimeoutException {
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost(customProperties.getHost());
        connectionFactory.setPort(Integer.getInteger(customProperties.getPort()));
        connectionFactory.setVirtualHost(customProperties.getVirtualHost());
        connectionFactory.setUsername(customProperties.getUsername());
        connectionFactory.setPassword(customProperties.getPassword());
        return connectionFactory.newConnection().createChannel();
    }

    @Bean
    public List<RabbitExchangeDefinition> getRabbitDefinition() {
        return rabbitCustomProperties.getRabbitDefinition();
    }

    /**
     * 创建交换机
     */
    private void createMqExchange(List<RabbitExchangeDefinition> rabbitDefinition) throws InvocationTargetException, IllegalAccessException {
        for (RabbitExchangeDefinition definition : rabbitDefinition) {
            boolean flag = false;
            String exchangeName = definition.getExchangeName();
            String exchangeType = definition.getExchangeType();
            Boolean delayExchange = definition.getDelayExchange();
            Class<RabbitMqConfig> clazz = RabbitMqConfig.class;
            Method[] methods = clazz.getDeclaredMethods();
            for (Method method : methods) {
                if (method.getName().equals("exchange" + capitalizeTheFirstLetter(exchangeType))) {
                    method.setAccessible(true);
                    method.invoke(this, exchangeName, delayExchange, exchangeType);
                    flag = true;
                    break;
                }
            }
            if (!flag) {
                log.error("未找到对应交换机类型：{}", exchangeType);
                flag = true;
            }
        }
    }

    /**
     * 声明队列和绑定
     */
    private void declareQueueAndBinding(List<RabbitExchangeDefinition> rabbitDefinition) throws InvocationTargetException, IllegalAccessException {
        for (RabbitExchangeDefinition definition : rabbitDefinition) {
            boolean flag = false;
            String exchangeName = definition.getExchangeName();
            String exchangeType = definition.getExchangeType();
            Boolean delayExchange = definition.getDelayExchange();
            List<RabbitQueueDefinition> rabbitQueueDefinitions = definition.getQueueDefinitions();
            Class<RabbitMqConfig> clazz = RabbitMqConfig.class;
            Method[] methods = clazz.getDeclaredMethods();
            for (Method method : methods) {
                if (method.getName().equals("queue" + capitalizeTheFirstLetter(exchangeType))) {
                    method.setAccessible(true);
                    method.invoke(this, rabbitQueueDefinitions, exchangeName, delayExchange);
                    flag = true;
                    break;
                }
            }
            if (!flag) {
                log.error("未找到对应交换机类型：{}", exchangeType);
                flag = true;
            }
        }
    }

    private void declareQueue(String queueName) {
        this.declareQueue(queueName, false, null, null);
    }

    /**
     * @Description 声明所有交换机
     */
    private void declareQueue(String queueName, Boolean isDelay, String exchangeName, String routing) {
        log.info("声明队列：{}", queueName);
        Queue queue = null;
        if (isDelay) {
            Map<String, Object> args = new HashMap<>();
            args.put("x-dead-letter-exchange", exchangeName + DELAY);
            args.put("x-dead-letter-routing-key", routing + DELAY);
            queue = QueueBuilder.durable(queueName).withArguments(args).build();
        } else {
            queue = QueueBuilder.durable(queueName).build();
        }
        rabbitMqRegister.registerBean(queueName, queue);
    }

    private String capitalizeTheFirstLetter(String str) {
        char firstChar = str.toUpperCase().charAt(0);
        String nextStr = str.substring(1);
        return firstChar + nextStr;
    }

}
